package org.ala.tree.lsm.sstable;

import java.io.File;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.MappedByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.channels.FileChannel.MapMode;
import java.nio.file.Path;
import java.nio.file.StandardOpenOption;
import java.util.ArrayList;
import java.util.List;

import org.ala.tree.lsm.LSMConfig;
import org.ala.tree.lsm.sstable.block.DataBlock;
import org.ala.tree.lsm.sstable.block.IndexBlock;
import org.ala.tree.lsm.sstable.block.MetaBlock;
import org.ala.tree.lsm.sstable.block.TailBlock;
import org.ala.tree.lsm.tools.FileTools;


/**
 * 把磁盘中的SSTable文件，映射成SStable对象
 *
 * @author ala
 * @date 2025年5月30日
 */
public class SSTableReader<V> {

	
	/**
	 * 	读取本层的内存映射，并按写入时间降序
	 * 	<p>	只会加载布隆过滤器
	 * 	<p>	如果要加载index，需要单独操作
	 */
	public List<SSTable<V>> read(LSMConfig config, int level, Class<V> vclass) throws IOException {
		String dir = String.format("%s/%s%d", config.ROOT_DIR, config.levelPrefix, level);
		File f = new File(dir);
		if (!f.exists()) {return null;}
		List<Path> paths = FileTools.filesAll(dir, "sstable");
		
		List<SSTable<V>> sstables = new ArrayList<>(paths.size());
		paths.forEach(p -> {
			try {
				sstables.add(simpleLoad(level, p, vclass));
			} catch (IOException e) {
				e.printStackTrace();
			}
		});
		sstables.sort((s1, s2) -> Long.compare(s2.getTimestamp(), s1.getTimestamp()));
		return sstables;
	}
	
	/**
	 * 	加载索引区
	 * 	@throws IOException 
	 */
	public void loadIndex(LSMConfig config, SSTable<V> sstable) throws IOException {
		String path = String.format("%s/%s%d/%s.sstable", config.ROOT_DIR, config.levelPrefix, sstable.getLevel(), sstable.getId());
		FileChannel fileChannel = FileChannel.open(Path.of(path), StandardOpenOption.READ);
		
		try {
			MappedByteBuffer buf = fileChannel.map(MapMode.READ_ONLY, sstable.getTb().getIndexOffset(), sstable.getTb().getIndexLen());
			IndexBlock<V> ib = IndexBlock.decode(buf);
			sstable.setIndex(ib.keyOffsetIndex());
		} finally {
			fileChannel.close();
		}
	}
	
	/**
	 * 	初步加载映射
	 * 	<p>	只加载布隆过滤器
	 * @throws IOException 
	 */
	public SSTable<V> simpleLoad(int level, Path path, Class<V> vclass) throws IOException {
		//	解析文件名
		String id = path.getFileName().toString().split("\\.")[0];
		
		FileChannel fileChannel = FileChannel.open(path, StandardOpenOption.READ);
		try {
			//	文件尾
			TailBlock tb = readTail(fileChannel);
			//	读元区，并且加载布隆过滤器
			MetaBlock mb = readMeta(fileChannel, tb, true);
			
			SSTable<V> sstable = new SSTable<>();
			sstable.setId(id);
			sstable.setLevel(level);
			sstable.setTimestamp(mb.getTimestamp());
			sstable.setBloomFilter(mb.getBloomFilter());
			sstable.setTb(tb);
			sstable.setDataBlockCapacity(mb.getDataBlockCapacity());
			sstable.setVclass(vclass);
			return sstable;
		} finally {
			fileChannel.close();
		}
	}
	/**
	 * 	读文件尾
	 * 	@throws IOException 
	 */
	public TailBlock readTail(FileChannel fileChannel) throws IOException {
		MappedByteBuffer buffer = fileChannel.map(MapMode.READ_ONLY, fileChannel.size() - TailBlock.LEN, TailBlock.LEN);
		return TailBlock.decode(buffer);
	}
	/**
	 * 	读元区
	 * 	@throws IOException 
	 */
	public MetaBlock readMeta(FileChannel fileChannel, TailBlock tb, boolean loadBloomFilterBytes) throws IOException {
		MappedByteBuffer buffer = fileChannel.map(MapMode.READ_ONLY, tb.getMetaOffset(), tb.getMetaLen());
		return MetaBlock.decode(buffer, loadBloomFilterBytes);
	}
	/**
	 * 	读索引区
	 * @throws IOException 
	 */
	public IndexBlock<V> readIndex(FileChannel fileChannel, TailBlock tb) throws IOException {
		MappedByteBuffer buffer = fileChannel.map(MapMode.READ_ONLY, tb.getIndexOffset(), tb.getIndexLen());
		return IndexBlock.decode(buffer);
	}
	/**
	 * 	读数据区
	 * 	@throws IOException 
	 */
	public DataBlock<V> readData(FileChannel fileChannel, int offset, int dataBlockCapacity, Class<V> vclass) throws IOException {
		//	读数据区长度
		ByteBuffer lenBuf = ByteBuffer.allocate(4);
		fileChannel.read(lenBuf);
		int len = lenBuf.getInt();
		//	读取DataBlock
		MappedByteBuffer buf = fileChannel.map(MapMode.READ_ONLY, offset + 4, len);
		DataBlock<V> dataBlock = DataBlock.decode(dataBlockCapacity, buf, vclass);
		return dataBlock;
	}
}
